package com.gitee.wsl.jvm.task.thread

//package io.github.doodle;

import com.gitee.wsl.api.TimeoutListener
import com.gitee.wsl.jvm.task.thread.Scheduler.TASK_POOL_MAP
import com.gitee.wsl.jvm.task.thread.executor.PipeExecutor
import com.gitee.wsl.jvm.task.thread.executor.TagExecutor
import com.gitee.wsl.jvm.task.thread.executor.ThreadPoolExecutorCount
import com.gitee.wsl.jvm.task.thread.executor.ThreadPoolExecutorCount.Companion.TYPE_CACHED
import com.gitee.wsl.jvm.task.thread.executor.ThreadPoolExecutorCount.Companion.TYPE_CPU
import com.gitee.wsl.jvm.task.thread.executor.ThreadPoolExecutorCount.Companion.TYPE_IO
import com.gitee.wsl.jvm.task.thread.executor.ThreadPoolExecutorCount.Companion.TYPE_SINGLE
import kotlinx.coroutines.Runnable
import timber.log.Timber
import java.util.Timer
import java.util.TimerTask
import java.util.concurrent.CompletableFuture
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.Executor
import java.util.concurrent.ExecutorService
import java.util.concurrent.Executors
import java.util.concurrent.ScheduledExecutorService
import java.util.concurrent.ScheduledFuture
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicInteger


val CPU_COUNT = Runtime.getRuntime().availableProcessors()

/**
 * Loading tasks scheduler.
 * For handling concurrency and priority.
 */
object Scheduler {

    private val WINDOW_SIZE = 2.coerceAtLeast(CPU_COUNT).coerceAtMost(4)

    val TASK_POOL_MAP: MutableMap<Task<*>, ExecutorService> = ConcurrentHashMap()

    val pipeExecutor: PipeExecutor by lazy { PipeExecutor(WINDOW_SIZE, WINDOW_SIZE * 2) }

    val tagExecutor: TagExecutor by lazy { TagExecutor() }

    // Windows size is one, tasks execute in serial.
    // Use to storage result bitmap.
    val storageExecutor: PipeExecutor by lazy { PipeExecutor(1) }

    val singlePoolExecutor: ExecutorService by lazy { TYPE_SINGLE(Thread.NORM_PRIORITY) }

    val cachedPool: ExecutorService by lazy { TYPE_CACHED(Thread.NORM_PRIORITY) }

    val cpuPool: ExecutorService by lazy { TYPE_CPU(Thread.NORM_PRIORITY) }

    val ioPool: ExecutorService by lazy { TYPE_IO(Thread.NORM_PRIORITY) }

    val scheduledExecutor: ScheduledExecutorService by lazy { Executors.newSingleThreadScheduledExecutor() }

    var executor: ExecutorService = cpuPool


    fun <T> execute(
        pool: ExecutorService = executor,
        //delay: Long=0,
        period: Int = Thread.NORM_PRIORITY,
        task: Task<T>
    ) {
        synchronized(TASK_POOL_MAP) {
            if (TASK_POOL_MAP[task] != null) {
                Timber.tag("ThreadUtils").e("Task can only be executed once.")
                return
            }
            TASK_POOL_MAP.put(task, pool)
        }

        pool.execute(task)

        /*if (period == 0L) {
            if (delay == 0L) {
                pool.execute(task)
            } else {
                val timerTask: TimerTask = object : TimerTask() {
                    override fun run() {
                        pool.execute(task)
                    }
                }
                TIMER.schedule(timerTask, unit.toMillis(delay))
            }
        } else {
            task.setSchedule(true)
            val timerTask: TimerTask = object : TimerTask() {
                override fun run() {
                    pool.execute(task)
                }
            }
            TIMER.scheduleAtFixedRate(timerTask, unit.toMillis(delay), unit.toMillis(period))
        }*/
    }

    /**
     * 运行或异常
     */
    fun runOrThrow(task: () -> Unit) {
        try {
            task()
        } catch (e: Throwable) {
            if (e is java.lang.RuntimeException) {
                throw e
            } else {
                throw java.lang.RuntimeException(e)
            }
        }
    }

    /**
     * 运行并吃掉异常
     */
    fun runAndTry(task: () -> Unit) {
        try {
            task()
        } catch (e: Throwable) {
            //略过
        }
    }

    /**
     * 并行执行
     */
    /*fun parallel(task: Runnable?): Future<*> {
        return parallelExecutor.submit(task)
    }*/

    /**
     * 并行执行
     */
    /* fun <T> parallel(task: Callable<T>): Future<T> {
        return parallelExecutor.submit(task)
    }*/

    /**
     * 异步执行
     */
    fun async(task: Runnable): CompletableFuture<Void> {
        return CompletableFuture.runAsync(task, executor)
    }

    /**
     * 异步执行
     */
    /*fun <U> async(task: Supplier<U>): CompletableFuture<U> {
        return CompletableFuture.supplyAsync<U>(task, asyncExecutor)
    }*/

    fun asyncAndTry(task: () -> Unit): CompletableFuture<Void> {
        return CompletableFuture.runAsync({ runAndTry(task) }, executor)
    }

    /**
     * 延迟执行
     */
    fun delay(task: Runnable, millis: Long): ScheduledFuture<*> {
        return scheduledExecutor.schedule(task, millis, TimeUnit.MILLISECONDS)
    }

    /**
     * 延迟执行并重复
     */
    fun delayAndRepeat(task: Runnable, millis: Long): ScheduledFuture<*> {
        return scheduledExecutor.scheduleWithFixedDelay(task, 1000, millis, TimeUnit.MILLISECONDS)
    }

    /**
     * 定时任务
     */
    fun scheduleAtFixedRate(
        task: Runnable,
        initialDelay: Long,
        millisPeriod: Long
    ): ScheduledFuture<*> {
        return scheduledExecutor.scheduleAtFixedRate(
            task,
            initialDelay,
            millisPeriod,
            TimeUnit.MILLISECONDS
        )
    }

    /**
     * 定时任务
     */
    fun scheduleWithFixedDelay(
        task: Runnable?,
        initialDelay: Long,
        millisDelay: Long
    ): ScheduledFuture<*> {
        return scheduledExecutor.scheduleWithFixedDelay(
            task,
            initialDelay,
            millisDelay,
            TimeUnit.MILLISECONDS
        )
    }


    /**
     * Cancel the given task.
     *
     * @param task The task to cancel.
     */
    fun cancel(task: Task<*>) {
        task.cancel()
    }

    /**
     * Cancel the given tasks.
     *
     * @param tasks The tasks to cancel.
     */
    fun cancel(vararg tasks: Task<*>) {
        if (tasks.isEmpty()) return
        for (task in tasks) {
            task.cancel()
        }
    }

    /**
     * Cancel the given tasks.
     *
     * @param tasks The tasks to cancel.
     */
    fun cancel(tasks: List<Task<*>>) {
        if (tasks.isEmpty()) return
        for (task in tasks) {
            task.cancel()
        }
    }

    /**
     * Cancel the tasks in pool.
     *
     * @param executorService The pool.
     */
    fun cancel(executorService: ExecutorService) {
        if (executorService is ThreadPoolExecutorCount) {
            TASK_POOL_MAP.forEach {
                if (it.value == executorService)
                    cancel(it.key)
            }
        } else {
            Timber.tag("ThreadUtils").e("The executorService is not ThreadUtils's pool.")
        }
    }


    fun shutdownAndAwaitTermination(pool: ExecutorService = executor, shutdownSeconds: Long = 5) {
        pool.shutdownNow()
        try {
            if (!pool.awaitTermination(shutdownSeconds, TimeUnit.SECONDS)) {
                pool.shutdownNow()
                if (!pool.awaitTermination(shutdownSeconds, TimeUnit.SECONDS)) {
                    throw RuntimeException("Failed to shutdown")
                }
            }
        } catch (ie: InterruptedException) {
            pool.shutdownNow()
            Thread.currentThread().interrupt()
            throw RuntimeException(ie)
        }
    }
}

abstract class Wrapper(val task: Runnable) : Runnable

abstract class SimpleTask<T> : Task<T>() {
    override fun onCancel() {
        Timber.tag("ThreadUtils").e("onCancel: " + Thread.currentThread())
    }

    override fun onFail(t: Throwable?) {
        Timber.tag("ThreadUtils").e(t, "onFail: ")
    }
}

abstract class Task<T> : Runnable {
    private val state = AtomicInteger(NEW)

    @Volatile
    private var isSchedule = false

    @Volatile
    private var runner: Thread? = null
    private var mTimer: Timer? = null
    private var mTimeoutMillis: Long = 0
    private var mTimeoutListener: TimeoutListener? = null
    private var deliver: Executor? = null

    @Throws(Throwable::class)
    abstract fun doInBackground(): T
    abstract fun onSuccess(result: T)
    abstract fun onCancel()
    abstract fun onFail(t: Throwable?)

    override fun run() {
        if (isSchedule) {
            if (runner == null) {
                if (!state.compareAndSet(NEW, RUNNING)) return
                runner = Thread.currentThread()
                if (mTimeoutListener != null) {
                    Timber.tag("ThreadUtils").w("Scheduled task doesn't support timeout.")
                }
            } else {
                if (state.get() != RUNNING) return
            }
        } else {
            if (!state.compareAndSet(NEW, RUNNING)) return
            runner = Thread.currentThread()
            if (mTimeoutListener != null) {
                mTimer = Timer()
                mTimer!!.schedule(object : TimerTask() {
                    override fun run() {
                        if (!isDone && mTimeoutListener != null) {
                            timeout()
                            mTimeoutListener!!.onTimeout()
                            onDone()
                        }
                    }
                }, mTimeoutMillis)
            }
        }
        try {
            val result = doInBackground()
            if (isSchedule) {
                if (state.get() != RUNNING) return
                getDeliver().execute { onSuccess(result) }
            } else {
                if (!state.compareAndSet(RUNNING, COMPLETING)) return
                getDeliver().execute {
                    onSuccess(result)
                    onDone()
                }
            }
        } catch (ignore: InterruptedException) {
            state.compareAndSet(CANCELLED, INTERRUPTED)
        } catch (throwable: Throwable) {
            if (!state.compareAndSet(RUNNING, EXCEPTIONAL)) return
            getDeliver().execute {
                onFail(throwable)
                onDone()
            }
        }
    }

    @JvmOverloads
    fun cancel(mayInterruptIfRunning: Boolean = true) {
        synchronized(state) {
            if (state.get() > RUNNING) return
            state.set(CANCELLED)
        }
        if (mayInterruptIfRunning) {
            if (runner != null) {
                runner!!.interrupt()
            }
        }
        getDeliver().execute {
            onCancel()
            onDone()
        }
    }

    private fun timeout() {
        synchronized(state) {
            if (state.get() > RUNNING) return
            state.set(TIMEOUT)
        }
        if (runner != null) {
            runner!!.interrupt()
        }
    }

    val isCanceled: Boolean
        get() = state.get() >= CANCELLED
    val isDone: Boolean
        get() = state.get() > RUNNING

    fun setDeliver(deliver: Executor): Task<T> {
        this.deliver = deliver
        return this
    }

    /**
     * Scheduled task doesn't support timeout.
     */
    fun setTimeout(timeoutMillis: Long, listener: TimeoutListener?): Task<T> {
        mTimeoutMillis = timeoutMillis
        mTimeoutListener = listener
        return this
    }

    fun setSchedule(isSchedule: Boolean) {
        this.isSchedule = isSchedule
    }

    private fun getDeliver(): Executor {
        return deliver!!
    }

    protected fun onDone() {
        TASK_POOL_MAP.remove(this)
        if (mTimer != null) {
            mTimer!!.cancel()
            mTimer = null
            mTimeoutListener = null
        }
    }

    companion object {
        private const val NEW = 0
        private const val RUNNING = 1
        private const val EXCEPTIONAL = 2
        private const val COMPLETING = 3
        private const val CANCELLED = 4
        private const val INTERRUPTED = 5
        private const val TIMEOUT = 6
    }
}

